package com.atguigu.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.TableProcess;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/2/4 12:04
 */
public class MyKafkaUtil {
    public static FlinkKafkaConsumer<String> getKafkaSource(String groupId,
                                                            String topic) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        props.setProperty("group.id", groupId);
        props.setProperty("auto.offset.reset", "latest");
        return new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);
    }
    
    // 写入固定的Topic
    public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
        //        return new FlinkKafkaProducer<>("hadoop162:9092,hadoop163:9092,hadoop164:9092", topic, new SimpleStringSchema());
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        properties.setProperty("transaction.timeout.ms", 1000 * 60 * 15 + "");
        return new FlinkKafkaProducer<String>(
            topic,
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {
                    return new ProducerRecord<>(topic, null, element.getBytes());
                }
            },
            properties,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    // 根据内容动态的写入不同的kafka Topic
    public static FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "hadoop162:9092,hadoop163:9092,hadoop164:9092");
        //如果15分钟没有更新状态，则超时 默认1分钟
        props.setProperty("transaction.timeout.ms", 1000 * 60 * 15 + "");
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default_topic",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element, @Nullable Long timestamp) {
                    final String topic = element.f1.getSinkTable();
                    final JSONObject data = element.f0.getJSONObject("data");
                    
                    return new ProducerRecord<>(topic, data.toJSONString().getBytes());
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }
}
